package com.larry.spark.test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


/**
 * 编号	字段名称	字段类型	字段含义
1	date	String	用户点击行为的日期
2	user_id	Long	用户的ID
3	session_id	String	Session的ID
4	page_id	Long	某个页面的ID
5	action_time	String	动作的时间点
6	search_keyword	String	用户搜索的关键词
7	click_category_id	Long	某一个商品品类的ID
8	click_product_id	Long	某一个商品的ID
9	order_category_ids	String	一次订单中所有品类的ID集合
10	order_product_ids	String	一次订单中所有商品的ID集合
11	pay_category_ids	String	一次支付中所有品类的ID集合
12	pay_product_ids	String	一次支付中所有商品的ID集合
13	city_id	Long	城市 id

 */
object Top10_1 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
    val sc = new SparkContext(conf)

    val dataLine = sc.textFile("data/user_visit_action.txt")
    //缓存
//    dataLine.cache()

    //2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
    //点击，下单，支付
    //点击数据
    val clickData: RDD[(String, Int)] = dataLine.filter(
      line => {
        val lines = line.split("_")
        val click = lines(6)
        click != "-1"
      }
    ).map(
      line => {
        val lines = line.split("_")
        val click = lines(6)
        (click, 1)
      }
    ).reduceByKey(_ + _)

    //下单数据
    val orderData = dataLine.filter(
      line => {
        val lines = line.split("_")
        val order = lines(8)
        order != "null"
      }
    ).flatMap(
      line => {
        val lines = line.split("_")
        val order = lines(8)
        val orders = order.split(",")
        orders.map((_,1))
      }
    ).reduceByKey(_+_)

    //支付数据
    val payData = dataLine.filter(
      line => {
        val lines = line.split("_")
        val order = lines(10)
        order != "null"
      }
    ).flatMap(
      line => {
        val lines = line.split("_")
        val order = lines(10)
        val orders = order.split(",")
        orders.map((_,1))
      }
    ).reduceByKey(_+_)

    //变为tuple后综合排序
    //转换格式(id,(点击)) => (id,(点击,0,0))
    val click = clickData.map {
      case (id, count) => {
        (id, (count, 0, 0))
      }
    }

    val order = orderData.map {
      case (id, count) => {
        (id, (0, count, 0))
      }
    }

    val pay = payData.map {
      case (id, count) => {
        (id, (0, 0, count))
      }
    }

    //连接数据
    val data = click.union(order).union(pay)

    //根据key聚合
    val datas = data.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )

    //排序 取前十
    val top = datas.sortBy(_._2, false).take(10)

    top.foreach(println)

    sc.stop()
  }

}
